package com.cmnit.analysis.dao

import com.cmnit.analysis.common.TDao
import org.apache.log4j.Logger
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.DataFrame

class GantryPlateDao extends TDao {
  private val logger: Logger = Logger.getLogger(this.getClass)

  /**
   * 广播门架信息码表
   */
  def broadcastGantryBase(): Broadcast[DataFrame] = {
    broadcastSQL(
      "select " +
        "gantryId," +
        "gantryType," +
        "'江西省' as province " +
        "from " +
        "dim.dim_tbl_gantrybaseinfo"
    )
  }

  /**
   * 广播车牌-车辆类型码表
   */
  def broadcastVehicleBase(): Broadcast[DataFrame] = {
    broadcastSQL(
      "select " +
        "vehiclePlate," +
        "vehicleType " +
        "from " +
        "dim.dim_tbl_vehiclebasicinfo"
    )
  }

  /**
   * 加工结果数据
   */
  def gantryPlateData(sqlTime: String, acctTime: String): DataFrame = {
    sparkSQL(
      "SELECT " +
        "t.province province, " +
        "t.gantryId gantryId, " +
        "t.gantryType gantryType, " +
        "NVL(b.vehicleType, '1') vehicleType, " +
        "COUNT(*) countNum, " +
        "'" + acctTime + "' as statisHour, " +
        "'" + acctTime.substring(0, 8) + "' as statisDay " +
        "FROM " +
        "(SELECT " +
        "DISTINCT " +
        "SUBSTR(a.gantryId, 1, 17) gantryId, " +
        "a.vehiclePlate, " +
        "b.province, " +
        "b.gantryType " +
        "FROM " +
        "ods.ods_etc_gantryvehdisdata a " +
        "JOIN " +
        "tmp_gantry_base_info b " +
        "ON a.gantryId = b.gantryId " +
        "WHERE "
        + sqlTime +
        ") t " +
        "LEFT JOIN " +
        "tmp_vehicle_base_info b " +
        "ON t.vehiclePlate = b.vehiclePlate " +
        "GROUP BY t.province,t.gantryId,t.gantryType,b.vehicleType "
    )
  }
}
